// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved.
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;
using System.Threading;
using System.Threading.Actors;
using System.Threading.Tasks;

namespace WordCount
{
    class ReduceActor : FabricAdHocActorBase, IDisposable
    {
        private int m_index;
        private int m_numMappers;
        private FabricActorClient m_client;

        public ReduceActor(int index, int numMappers)
            : base ("reducer" + index)
        {
            m_index = index;
            m_numMappers = numMappers;
        }

        public void Initialize(FabricActorClient client, string assemblyName, byte[] assemblyBytes)
        {
            m_client = client;
            client.AddAssembly(assemblyName, assemblyBytes);
            client.CallMethod("InitializeReducer", new object[] { m_index, m_numMappers });
        }

        public void Dispose()
        {
            if (m_client != null)
            {
                m_client.Dispose();
                m_client = null;
            }
        }


        [ActorMethod]
        public static object InitializeReducer(dynamic state, dynamic[] args)
        {
            state.index = (int)args[0];
            state.numMappers = (int)args[1];
            state.mappersCompleted = 0;

            state.InitializeDictionaryStuff();

            return null;
        }

        // Receives results from a mapper.
        [ActorMethod]
        public static object Reduce(dynamic state, dynamic[] args)
        {
            string filePath = args[0];
            string justTheName = Path.GetFileNameWithoutExtension(filePath);
            state.Publish("Update", justTheName, (int) state.index);
            BinaryFormatter bf = new BinaryFormatter();
            using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read))
            {
                var incomingDictionary = (Dictionary<string, int>)bf.Deserialize(stream);
                int totalWordCount = 0;
                foreach (var kvp in incomingDictionary)
                {
                    state.SetOrAdd(kvp.Key, kvp.Value);
                    totalWordCount++;

                    // Purposefully slow down the process by pausing every 10 words
                    if ((totalWordCount % 10) == 0)
                    {
                        //Thread.Sleep(1);
                        int elapsedTime = Environment.TickCount;
                        int targetTime = elapsedTime + 2;
                        int newTime = elapsedTime;
                        while (newTime < targetTime)
                        {
                            newTime = Environment.TickCount;
                        }
                    }
                    if ((totalWordCount % 100) == 0)
                    {
                        state.Publish("Update", String.Format("{0}\n{1}", justTheName, totalWordCount), (int)state.index);
                    }
                }
            }

            state.Publish("Update", "<idle>", (int) state.index);
            return null;
        }


        // A mapper has completed mapping.  Once all mappers have completed mapping,
        // we can consider our reduction to complete, aggregate our results
        // and ship them to the aggregator.
        [ActorMethod]
        public static object MappingComplete(dynamic state, dynamic[] args)
        {
            var mappersCompleted = (int) state.mappersCompleted;
            var numMappers = (int)state.numMappers;

            mappersCompleted += 1;
            state.mappersCompleted = mappersCompleted;

            if (mappersCompleted == numMappers)
            {
                string outdir = Environment.GetFolderPath(Environment.SpecialFolder.MyDocuments) + "\\MRFiles";
                string fname = String.Format("{0}\\{1}-from-{2}", outdir, Guid.NewGuid().ToString(), state.index);
                BinaryFormatter bf = new BinaryFormatter();

                using (var outstream = new FileStream(fname, FileMode.Create, FileAccess.Write))
                {
                    var list = (List<KeyValuePair<string, int>>)state.EnumerateAll();
                    bf.Serialize(outstream, list);
                    outstream.Flush();
                }
                var proxy = state.GetActorProxy("fabric:/actor/adhoc/aggregator");
                proxy.ProcessAggregatedInfo(fname);

                proxy.ReducingComplete(); // Probably unnecessary now...

                state.Publish("Update", "Done", (int)state.index);
            }

            return null;
        }

        //
        // Experimental Dictionary logic
        // Supports SetOrAdd and efficient enumeration that does not require touching all state
        // items via GetAllKeysWithPrefix.
        //

        // Initializes
        [ActorMethod]
        public static object InitializeDictionaryStuff(IActorState state, object[] args)
        {
            state.Remove("finalKey");
            state.Set("numElements", 0);

            // Remove all old keys, if any
            var oldKeys = state.GetAllKeysWithPrefix("$$");
            foreach (var key in oldKeys) state.Remove("$$" + key);

            return null;
        }

        [ActorMethod]
        public static object SetOrAdd(dynamic state, dynamic[] args)
        {
            string key = args[0];
            int value = args[1];

            bool added = false;
            string searchKey = "$$" + key;
            if (state.Contains(searchKey))
            {
                var oldTuple = (Tuple<int, string>)state[searchKey];
                var newTuple = Tuple.Create(oldTuple.Item1 + value, oldTuple.Item2);
                state[searchKey] = newTuple;
            }
            else
            {
                added = true;
                var numElements = (int)state.numElements;
                if (numElements == 0)
                {
                    var newTuple = Tuple.Create(value, (string)null);
                    state.finalKey = searchKey;
                    state.numElements = 1;
                    state[searchKey] = newTuple;
                }
                else
                {
                    var oldFinalKey = (string)state.finalKey;
                    var newTuple = Tuple.Create(value, oldFinalKey);
                    state.finalKey = searchKey;
                    state.numElements = numElements + 1;
                    state[searchKey] = newTuple;
                }
            }

            return added;
        }

        [ActorMethod]
        public static object EnumerateAll(dynamic state, dynamic[] args)
        {
            List<KeyValuePair<string, int>> result = new List<KeyValuePair<string, int>>();
            if ((int)state.numElements > 0)
            {
                var key = state.finalKey as string;
                while (key != null)
                {
                    var tuple = (Tuple<int, string>)state[key];
                    result.Add(new KeyValuePair<string, int>(key.Substring(2), tuple.Item1));
                    key = tuple.Item2;
                }
            }
            return result;
        }
    }
}
